package net.wlm.others;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.wlm.utils.JdbcUtil;
import net.wlm.utils.KafkaUtil;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

public class GongDanVehicleJob {
    public static void main(String[] args) throws Exception {
        // 执行环境-env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 数据源-source
        DataStream<String> stringDataStream = KafkaUtil.consumerKafka(env, "vehicle-data");

        // 数据转换-transformation
        // 数据输出-sink
        ProcessLog(stringDataStream);


        // 触发执行-execute
        env.execute("GongDanVehicleJob");


    }

    private static void ProcessLog(DataStream<String> stringDataStream) {
        SingleOutputStreamOperator<Tuple2<String, String>> mapStream = stringDataStream.map(new MapFunction<String, Tuple2<String, String>>() {
            @Override
            public Tuple2<String, String> map(String value) throws Exception {

                JSONObject jsonObject = JSON.parseObject(value);
                String vin = jsonObject.getString("vin");
                String geoHash = GeoHashUdf.getGeoHash(jsonObject.getDouble("latitude"), jsonObject.getDouble("longitude"));
                return Tuple2.of(geoHash, vin);
            }
        });

        SingleOutputStreamOperator<String> vinStateStream = mapStream.keyBy(t -> t.f0).process(new ProcessFunction<Tuple2<String, String>, String>() {

            private transient MapState<String,Boolean> vinSet;

            @Override
            public void open(Configuration parameters) throws Exception {
                MapStateDescriptor<String, Boolean> descriptor =
                        new MapStateDescriptor<>("vin_map", String.class, Boolean.class);
                vinSet = getRuntimeContext().getMapState(descriptor);
            }

            @Override
            public void processElement(Tuple2<String, String> value, ProcessFunction<Tuple2<String, String>, String>.Context ctx, Collector<String> out) throws Exception {
                String geoHash = value.f0;
                String vin = value.f1;

                // 添加或更新VIN
                vinSet.put(vin, true);

                // 获取当前数量
                int count = 0;
                for (String key : vinSet.keys()) {
                    count++;
                }
                JSONObject jsonObject = new JSONObject();
                    jsonObject.put("geoHash", geoHash);
                    jsonObject.put("vinCount", count);
                    out.collect(jsonObject.toJSONString());
            }
        });

        String[] columns = {"geoHash", "vinCount"};
        String upsertSql = "INSERT INTO vehicle_real.vehicle_geohash_vin_count(\n" +
                "    geoHash, vinCount\n" +
                ") VALUES (\n" +
                "    ?, ?\n" +
                ")";
                JdbcUtil.sinkToClickhouseUpsertJson(vinStateStream, columns, upsertSql);

    }


}
